package com.rrd.dw.mr.gzip.kafka;

import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.MD5Util;
import com.ucredit.babel.common.kafka.CrawlResult;
import com.ucredit.babel.common.model.BaseModel;
import com.ucredit.babel.common.model.Job;
import com.weicai.caesar.CaesarUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.mapreduce.Mapper.Context;

import java.io.IOException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.Random;
import java.util.regex.Pattern;


public abstract class ParseBase {
    public static final String CRT001 = "\001";
    public static final Pattern CRT001_PATTERN = Pattern.compile("\001");
    private StringBuilder sb = new StringBuilder(500);
    private StringBuilder txSb = new StringBuilder(500);
    private String partition = "";
    private Integer fileNum = 1;

    protected StringBuilder getBaseInfo(BaseModel<?> obj) {
        Job job = obj.getJob();
        txSb.delete(0, txSb.length());
        txSb.append(job.getRoute() == null ? "" : job.getRoute()).append(CRT001);
        txSb.append(job.getSubject() == null ? "" : job.getSubject()).append(CRT001);
        txSb.append(job.getChannel() == null ? "" : job.getChannel()).append(CRT001);
        return txSb;
    }

    protected StringBuilder getBaseInfo(BaseModel<?> obj, String successful) {
        Job job = obj.getJob();
        CrawlResult result = job.getResult();
        if (result == null) {
            result = new CrawlResult();
        }
        sb.delete(0, sb.length());
        String timestamstr = obj.getTimestamp() == null ? "" : DateUtil.formateDateStr(new Date(obj.getTimestamp()));
        String idNo = StringUtils.isBlank(obj.getIdNumber()) ? "" : CaesarUtil.isEncrypted(obj.getIdNumber()) ? obj.getIdNumber() : CaesarUtil.encode(obj.getIdNumber());
        String userKey = obj.getUserKey() == null ? "" : obj.getUserKey();
        String auditId = obj.getLoanKey() == null ? "" : obj.getLoanKey();
        String job_id = job.getJobID() == null ? "" : job.getJobID();
        String id = MD5Util.getMd5(job_id + timestamstr + idNo + userKey + auditId);

        sb.append(id).append(CRT001);
        sb.append(idNo).append(CRT001);
        sb.append(userKey).append(CRT001);
        sb.append(auditId).append(CRT001);
        sb.append(job_id).append(CRT001);
        sb.append(timestamstr).append(CRT001);

        //是否爬取完成
        sb.append(job.isFinished() == null ? "" : job.isFinished()).append(CRT001);
        //是否有请求
        sb.append(job.isRequested() == null ? "" : job.isRequested()).append(CRT001);
        //是否爬取成功
        sb.append(job.isSuccessful() == null ? "" : job.isSuccessful()).append(CRT001);
        //爬取开始时间
        sb.append(job.getStartTime() == null ? "" : DateUtil.formateDateStr(job.getStartTime())).append(CRT001);
        //爬取结束时间
        sb.append(job.getEndTime() == null ? "" : DateUtil.formateDateStr(job.getEndTime())).append(CRT001);
        //数据爬取业务系统
        sb.append(job.getSystemID() == null ? "" : job.getSystemID()).append(CRT001);
        //手机号码
        sb.append(job.getQuery() == null ? "" : job.getQuery()).append(CRT001);
        //重试次数
        sb.append(job.getTryTimes() == null ? "" : job.getTryTimes()).append(CRT001);
        //JOB类型
        sb.append(obj.getType() == null ? "" : obj.getType().name()).append(CRT001);
        //爬取结果消息
        sb.append(result.getMessage() == null ? "" : result.getMessage()).append(CRT001);
        //消息code
        sb.append(result.getCode() == null ? "" : result.getCode()).append(CRT001);

        Random ra =new Random();
        if (fileNum > 1) {
            setPartition("," + ra.nextInt(fileNum));
        }
        return sb;
    }

    protected Object formatValue(Object value) {
        Object ret;
        if (value instanceof String) {
            ret = ((String) value).replace(CRT001, "");
        } else {
            if (value == null){
                ret = "";
            } else {
                ret = value;
            }
        }
        return ret;
    }

    public void setPartition(String p) {
        partition = p;
    }

    public String getPartition() {
        return partition;
    }

    public String getKey() {
        return getType() + getSubType() + getPartition();
    }

    public String getSubType() {
        return "";
    }

    public void setFileNum(Integer num) {
        fileNum = num;
    }

    public abstract boolean parseJob(Context context, BaseModel<?> model, String successful) throws IOException, InterruptedException;

    public abstract String getTableName();

    public abstract String getType();

    public abstract String getComment();
}
